/*
* File Name:msg_bus.c
* Version:v1.0.2
* Change Logs:
* Date             Author          main change
* 2024-01-02       lanxin          the first version
* 2024-01-21       lanxin          rename to msg_bus.c
*                                  recode
* 2024-01-23       lanxin          fix publish err
* 2024-01-24       lanxin          change _msg_publish api
* 2024-01-25       lanxin          fix subscribe faile
* 2024-03-02       lanxin          update api function
* 2024-03-05       lanxin          add xx_urgent api
* 2024-08-23       lanxin          improve efficiency
* 2025-03-12       lanxin          make rt_mq_send return result adapt to rtt version
*
*/

/********************************************* READ ME BEGIN *************************************************************/

/* Brief:
*
* *1.基于 RT_Thread message queue 的订阅与发布的消息通道。
* - 支持消息的发布
* - 支持回调函数的发布
*
*/

/* Notice:
*
* *1.建议使用 publish_by_node,这个相对于 publish_by_name 不用查找主题，效率更高。
*
* *2.几乎每个 api 里面都有 init 状态的检测，防止未进行初始化。
* - 当然这个会多出来一点点的消耗，如果要关闭的话，注释掉 CHECK_INIT 即可。
* - 注释掉后需自行保证模块已经进行初始化。
*
* *3.消息和函数回调的发布均在调用 api 的线程里面，所以线程栈需要适当调整。
*
* *4.消息的接收需要自己在线程中进行接收， rt_mq_receive(mq); mq 为订阅主题传入的 mq 参数。
*
* *5.不能以同一个事件函数订阅多个主题。
*
*/

/********************************************* READ ME END ***************************************************************/

/* source file */
#include "msg_bus.h"

/* private define */

#define _MSG_LOCK()                                 if(mutex_msg_bus) rt_mutex_take(mutex_msg_bus,RT_WAITING_FOREVER);
#define _MSG_UNLOCK()                               if(mutex_msg_bus) rt_mutex_release(mutex_msg_bus);
#define FS_ASSERT(PARA)                             if((PARA) == 0) { rt_kprintf("file:%s",__FILE__);RT_ASSERT(0);}
#define FS_PARA_CHECK_NULL(PARA)                    if((PARA) == 0) {rt_kprintf("file:%s line:%d para is null\r\n",__FILE__,__LINE__); return;}
#define FS_PARA_CHECK_NULL_THEN_RETURN(PARA,VALUE)  if((PARA) == 0) {rt_kprintf("file:%s line:%d para is null\r\n",__FILE__,__LINE__); return VALUE;}

/* rtthread change rt_mq_send_xx return result,before 5.1.0 is rt_err_t,after that is size */
#if RTTHREAD_VERSION >= 51000
	#define MQ_RESULT_IS_SIZE 1
#else
	#define MQ_RESULT_IS_SIZE 0
#endif

//#define CHECK_INIT //是否在调用 api 是检测初始化状态

/* function declaration */

/* variable declaration */
/* 消息模式 */
enum
{
	E_MSG_BUS_MODE_MSG = 1,
	E_MSG_BUS_MODE_FUNCTION,
};

/* 订阅者参数 */
typedef struct subscribe_para
{
	uint8_t mode;

	void *cb_para;
	union
	{
		rt_mq_t mq;
		msg_bus_cb cb;
	} argv;
} s_subscribe_para_t;

/* 订阅者 */
typedef struct msg_bus_subscriber
{
	rt_list_t list_subscriber;//订阅者结点

	s_subscribe_para_t para;//订阅者参数
} s_msg_bus_subscriber_t;

/* 主题 */
typedef struct msg_bus_topic
{
	rt_list_t list_topic;//主题结点

	rt_list_t list_subscriber_function;
	rt_list_t list_subscriber_msg;

	char *topic_name;//主题名称

	uint8_t enable;
} s_msg_bus_topic_t;

static rt_list_t list_msg_bus_topic_head;//主题的头结点
static rt_mutex_t mutex_msg_bus = RT_NULL;
static uint8_t _msg_bus_init_state = 0;

static void show_msg_bus ( void )
{
	if ( rt_list_isempty ( &list_msg_bus_topic_head ) )
	{
		rt_kprintf ( "there is no topic\r\n" );
		return ;
	}

	_MSG_LOCK();

	rt_list_t *tmp = 0;
	s_msg_bus_topic_t *tmp_topic = 0;
	rt_kprintf ( "here is %d topic\r\n", rt_list_len ( &list_msg_bus_topic_head ) );

	rt_list_for_each ( tmp, &list_msg_bus_topic_head )
	{
		tmp_topic = rt_list_entry ( tmp, s_msg_bus_topic_t, list_topic );

		if ( tmp_topic == 0 )
			continue;

		rt_kprintf ( "topic %s has %d msg sub %d func sub,topic is %s\r\n", tmp_topic->topic_name,
		             rt_list_len ( ( &tmp_topic->list_subscriber_msg ) ),
		             rt_list_len ( ( &tmp_topic->list_subscriber_function ) ),
		             tmp_topic->enable ? "enable" : "disable" );
	}

	_MSG_UNLOCK();
}

static int msg_bus_cmd ( int argc, char **argv )
{
	if ( argc == 2 )
	{
		if ( rt_strcmp ( argv[1], "show" ) == 0 )
		{
			show_msg_bus();
			return 0;
		}
	}

	return 0;
}

static uint8_t _msg_bus_check_init ( void )
{
	/* 未初始化进行初始化 */
	if ( 0 == _msg_bus_init_state )
	{
		/* 初始化主题链表 */
		rt_list_init ( &list_msg_bus_topic_head );

		/* 创建互斥锁 */
		mutex_msg_bus = rt_mutex_create ( "mutex_msg_bus", RT_IPC_FLAG_PRIO );
		FS_ASSERT ( mutex_msg_bus != RT_NULL );

		/* 更新状态 */
		_msg_bus_init_state = 1;

		rt_kprintf ( "soft message bus initial down\r\n" );
	}

	return _msg_bus_init_state;
}

//发布消息
//list_topic 为创建主题时返回的结点
//订阅者工作模式为 消息型 msg len 为发送的消息内容和长度
//订阅者工作模式为 回调型 msg len 均填 0 就行，未使用
static int _msg_bus_topic_publish ( s_msg_bus_topic_t *topic, uint8_t mode, const void *msg, int len, uint8_t urgent )
{
	FS_PARA_CHECK_NULL_THEN_RETURN ( topic, -1 );

	if ( topic->enable == 0 )
		return -1;

	rt_list_t *_list_sub = mode == E_MSG_BUS_MODE_FUNCTION ? &topic->list_subscriber_function :
	                       mode == E_MSG_BUS_MODE_MSG ? &topic->list_subscriber_msg : 0;
	FS_PARA_CHECK_NULL_THEN_RETURN ( _list_sub, -1 );

	s_msg_bus_subscriber_t *_sub_temp = RT_NULL;

	rt_list_t *list_tmp = RT_NULL;
	/* 遍历所有的订阅者 */
	rt_list_for_each ( list_tmp, _list_sub )
	{
		_sub_temp = rt_list_entry ( list_tmp, s_msg_bus_subscriber_t, list_subscriber );

		if ( _sub_temp == 0 )
			return -1;
		else if ( mode == E_MSG_BUS_MODE_FUNCTION && _sub_temp->para.argv.cb )
			_sub_temp->para.argv.cb ( _sub_temp->para.cb_para );
		else if ( mode == E_MSG_BUS_MODE_MSG && _sub_temp->para.argv.mq )
		{
			FS_PARA_CHECK_NULL_THEN_RETURN ( msg, -1 );
			FS_PARA_CHECK_NULL_THEN_RETURN ( len, -1 );

			if ( urgent )
			{
#if MQ_RESULT_IS_SIZE

				if ( rt_mq_urgent ( _sub_temp->para.argv.mq, msg, len ) != len )
#else
				if ( rt_mq_urgent ( _sub_temp->para.argv.mq, msg, len ) != RT_EOK )
#endif
				{
					return -1;
				}
			}
			else
			{
#if MQ_RESULT_IS_SIZE

				if ( rt_mq_send ( _sub_temp->para.argv.mq, msg, len ) != len )
#else
				if ( rt_mq_send ( _sub_temp->para.argv.mq, msg, len ) != RT_EOK )
#endif
				{
					return -1;
				}
			}
		}
	}

	return 0;
}

//查找主题
static s_msg_bus_topic_t *_msg_bus_topic_find_node ( const char *topic )
{
	s_msg_bus_topic_t *_topic_find = RT_NULL;

	rt_list_t *list_tmp = 0;

	/* 遍历所有主题，并判断其主题名称 */
	rt_list_for_each ( list_tmp, &list_msg_bus_topic_head )
	{
		_topic_find = rt_list_entry ( list_tmp, s_msg_bus_topic_t, list_topic );

		if ( _topic_find && _topic_find->topic_name && rt_strcmp ( _topic_find->topic_name, topic ) == 0 )
			return _topic_find;
	}

	return 0;
}

//查找订阅者结点
static s_msg_bus_subscriber_t *_msg_bus_subscirbe_find_node ( const char *topic, s_subscribe_para_t para )
{
	/* 查找主题 */
	s_msg_bus_topic_t *_topic_temp = _msg_bus_topic_find_node ( topic );
	s_msg_bus_subscriber_t *_sub_temp = RT_NULL;

	FS_PARA_CHECK_NULL_THEN_RETURN ( _topic_temp, 0 );

	rt_list_t *_list_sub = para.mode == E_MSG_BUS_MODE_FUNCTION ? &_topic_temp->list_subscriber_function :
	                       para.mode == E_MSG_BUS_MODE_MSG ? &_topic_temp->list_subscriber_msg : 0;
	FS_PARA_CHECK_NULL_THEN_RETURN ( _list_sub, 0 );

	rt_list_t *list_tmp = 0;
	rt_list_for_each ( list_tmp, _list_sub )
	{
		_sub_temp = rt_list_entry ( list_tmp, s_msg_bus_subscriber_t, list_subscriber );

		if ( _sub_temp == 0 )
			return 0;
		else if ( para.mode == E_MSG_BUS_MODE_FUNCTION && para.argv.cb == _sub_temp->para.argv.cb )
			return _sub_temp;
		else if ( para.mode == E_MSG_BUS_MODE_MSG && para.argv.mq == _sub_temp->para.argv.mq )
			return _sub_temp;
	}

	return 0;
}

//解除订阅一个主题
static int _msg_bus_topic_unsubscribe ( const char *topic_name, s_subscribe_para_t para )
{
	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, -1 );

	int _res = -1;
	s_msg_bus_subscriber_t *_sub_temp = RT_NULL;

	_MSG_LOCK();

	/* 查找订阅者 */
	_sub_temp = _msg_bus_subscirbe_find_node ( topic_name, para );

	if ( _sub_temp )
	{
		/* 从链表移除并释放资源 */
		rt_list_remove ( &_sub_temp->list_subscriber );
		rt_free ( _sub_temp );

		_res = 0;
	}

	_MSG_UNLOCK();

	return _res;
}

//订阅主题
static int _msg_bus_subscribe ( const char *topic_name, s_subscribe_para_t para )
{
	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, -1 );

	_MSG_LOCK();

	/* 查找主题 */
	s_msg_bus_topic_t *_topic_temp = _msg_bus_topic_find_node ( topic_name );

	if ( 0 == _topic_temp )
	{
		_MSG_UNLOCK();

		return -1;
	}

	/* 查找订阅者 */
	s_msg_bus_subscriber_t *_sub_temp = _msg_bus_subscirbe_find_node ( topic_name, para );

	if ( _sub_temp )
	{
		_MSG_UNLOCK();
		return 0;
	}

	/* 创建一个新的订阅者 */
	s_msg_bus_subscriber_t *_new_sub = rt_calloc ( 1, sizeof ( s_msg_bus_subscriber_t ) );

	if ( _new_sub == 0 )
	{
		_MSG_UNLOCK();
		return 0;
	}

	/* 初始化订阅者结点并将其添加到该主题的订阅者链表内 */
	rt_list_init ( & ( _new_sub->list_subscriber ) );

	_new_sub->para.mode = para.mode;

	if ( para.mode == E_MSG_BUS_MODE_FUNCTION )
	{
		_new_sub->para.argv.cb = para.argv.cb;
		_new_sub->para.cb_para = para.cb_para;
		rt_list_insert_before ( &_topic_temp->list_subscriber_function, & ( _new_sub->list_subscriber ) );
	}
	else if ( para.mode == E_MSG_BUS_MODE_MSG )
	{
		_new_sub->para.argv.mq = para.argv.mq;
		rt_list_insert_before ( &_topic_temp->list_subscriber_msg, & ( _new_sub->list_subscriber ) );
	}
	else
	{
		rt_free ( _new_sub );
		_MSG_UNLOCK();

		return -1;
	}

	_MSG_UNLOCK();

	return 0;
}

//创建一个主题
//创建成功返回主题结点否则返回 0
rt_list_t *msg_bus_topic_create ( const char *topic_name )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, 0 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, 0 );

	/* 存在该主题则直接返回该主题结点 */
	/* 里面有互斥保护 */
	rt_list_t *_list_topic_temp = msg_bus_topic_find ( topic_name );

	if ( _list_topic_temp )
		return _list_topic_temp;

	_MSG_LOCK();

	/* 新建一个主题 */
	s_msg_bus_topic_t *_new_topic = rt_calloc ( 1, sizeof ( s_msg_bus_topic_t ) );

	if ( _new_topic == 0 )
	{
		_MSG_UNLOCK();
		return 0;
	}

	/* 开辟用于保存主题名称的空间 */
	_new_topic->topic_name = rt_calloc ( 1, rt_strlen ( topic_name ) + 1 );

	if ( _new_topic->topic_name == 0 )
	{
		rt_free ( _new_topic );
		_MSG_UNLOCK();
		return 0;
	}

	rt_snprintf ( _new_topic->topic_name, rt_strlen ( topic_name ) + 1, "%s", topic_name );

	/* 初始化结点并将本身添加到主题头结点上 */
	rt_list_init ( &_new_topic->list_topic );
	rt_list_insert_before ( &list_msg_bus_topic_head, & ( _new_topic->list_topic ) );

	/* 初始化订阅者结点 */
	rt_list_init ( &_new_topic->list_subscriber_function );
	rt_list_init ( &_new_topic->list_subscriber_msg );

	/* 默认使能主题 */
	_new_topic->enable = 1;

	_MSG_UNLOCK();

	/* 返回主题结点 */
	return &_new_topic->list_topic;
}

//通过主题名称找到主题结点
rt_list_t *msg_bus_topic_find ( const char *topic_name )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, 0 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, 0 );

	rt_list_t *_list_topic = RT_NULL;

	_MSG_LOCK();

	s_msg_bus_topic_t *_topic = _msg_bus_topic_find_node ( topic_name );

	if ( _topic )
		_list_topic = &_topic->list_topic;

	_MSG_UNLOCK();

	return _list_topic;
}

//订阅一个主题 消息模式
//mq 为接收该消息的消息队列
//主题不存在时，不创建该主题
int msg_bus_topic_subscribe ( const char *topic_name, rt_mq_t mq )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	s_subscribe_para_t para =
	{
		.mode = E_MSG_BUS_MODE_MSG,
		.argv.mq = mq,
	};

	/* 未找到主题时创建主题 */
	if ( msg_bus_topic_find ( topic_name ) == 0 )
	{
		rt_list_t *_topic = msg_bus_topic_create ( topic_name );
		FS_PARA_CHECK_NULL_THEN_RETURN ( _topic, -1 );
	}

	return _msg_bus_subscribe ( topic_name, para );
}

//订阅一个主题 事件模式
//cb 为接收到消息的回调函数
//主题不存在时，不创建该主题
int msg_bus_topic_subscribe_function ( const char *topic_name, msg_bus_cb cb, void *cb_para )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	s_subscribe_para_t para =
	{
		.mode = E_MSG_BUS_MODE_FUNCTION,
		.argv.cb = cb,
		.cb_para = cb_para,
	};

	/* 未找到主题时创建主题 */
	if ( msg_bus_topic_find ( topic_name ) == 0 )
	{
		rt_list_t *_topic = msg_bus_topic_create ( topic_name );
		FS_PARA_CHECK_NULL_THEN_RETURN ( _topic, -1 );
	}

	return _msg_bus_subscribe ( topic_name, para );
}

//解除订阅一个主题，订阅者工作模式为消息型
int msg_bus_topic_unsubscribe ( const char *topic_name, rt_mq_t mq )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( mq, -1 );

	/* 查找订阅者 */
	s_subscribe_para_t para =
	{
		.mode = E_MSG_BUS_MODE_MSG,
		.argv.mq = mq,
	};

	return _msg_bus_topic_unsubscribe ( topic_name, para );
}

//解除订阅一个主题，订阅者工作模式为回调函数型
int msg_bus_topic_unsubscribe_function ( const char *topic_name, msg_bus_cb cb )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( cb, -1 );

	/* 查找订阅者 */
	s_subscribe_para_t para =
	{
		.mode = E_MSG_BUS_MODE_FUNCTION,
		.argv.cb = cb,
	};

	return _msg_bus_topic_unsubscribe ( topic_name, para );
}

//通过主题结点发布消息
int msg_bus_publish_msg_by_node ( rt_list_t *list_topic, const void *msg, int len )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( list_topic, -1 );
	s_msg_bus_topic_t *_topic_temp = rt_list_entry ( list_topic, s_msg_bus_topic_t, list_topic );

	_MSG_LOCK();
	int _res = _msg_bus_topic_publish ( _topic_temp, E_MSG_BUS_MODE_MSG, msg, len, 0 );
	_MSG_UNLOCK();

	return _res;
}

//通过主题结点发布消息-紧急发布
int msg_bus_publish_msg_by_node_urgent ( rt_list_t *list_topic, const void *msg, int len )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( list_topic, -1 );
	s_msg_bus_topic_t *_topic_temp = rt_list_entry ( list_topic, s_msg_bus_topic_t, list_topic );

	_MSG_LOCK();
	int _res = _msg_bus_topic_publish ( _topic_temp, E_MSG_BUS_MODE_MSG, msg, len, 1 );
	_MSG_UNLOCK();

	return _res;
}

//通过主题结点调用回调函数
int msg_bus_publish_function_by_node ( rt_list_t *list_topic )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( list_topic, -1 );
	s_msg_bus_topic_t *_topic_temp = rt_list_entry ( list_topic, s_msg_bus_topic_t, list_topic );

	_MSG_LOCK();
	int _res = _msg_bus_topic_publish ( _topic_temp, E_MSG_BUS_MODE_FUNCTION, 0, 0, 0 );
	_MSG_UNLOCK();

	return _res;
}

//通过主题名称发布消息
int msg_bus_publish_msg_by_name ( const char *topic_name, const void *msg, int len )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, -1 );
	int _res = -1;

	_MSG_LOCK();

	s_msg_bus_topic_t *_topic_temp = _msg_bus_topic_find_node ( topic_name );

	if ( _topic_temp )
		_res = _msg_bus_topic_publish ( _topic_temp, E_MSG_BUS_MODE_MSG, msg, len, 0 );

	_MSG_UNLOCK();

	return _res;
}

//通过主题名称发布消息-紧急发布
int msg_bus_publish_msg_by_name_urgent ( const char *topic_name, const void *msg, int len )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, -1 );
	int _res = -1;

	_MSG_LOCK();

	s_msg_bus_topic_t *_topic_temp = _msg_bus_topic_find_node ( topic_name );

	if ( _topic_temp )
		_res = _msg_bus_topic_publish ( _topic_temp, E_MSG_BUS_MODE_MSG, msg, len, 1 );

	_MSG_UNLOCK();

	return _res;
}

//通过主题名称调用回调函数
int msg_bus_publish_function_by_name ( const char *topic_name )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, -1 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, -1 );

	int _res = -1;

	_MSG_LOCK();
	s_msg_bus_topic_t *_topic_temp = _msg_bus_topic_find_node ( topic_name );

	if ( _topic_temp )
		_res = _msg_bus_topic_publish ( _topic_temp, E_MSG_BUS_MODE_FUNCTION, 0, 0, 0 );

	_MSG_UNLOCK();

	return _res;
}

//设置主题的使能状态 0 失能 1 使能
int msg_bus_set_topic_enable ( const char *topic_name, uint8_t enable )
{
#ifdef CHECK_INIT
	uint8_t _init = _msg_bus_check_init();
	FS_PARA_CHECK_NULL_THEN_RETURN ( _init, 0 );
#endif

	FS_PARA_CHECK_NULL_THEN_RETURN ( topic_name, -1 );

	_MSG_LOCK();

	int _res = -1;

	s_msg_bus_topic_t *_topic = _msg_bus_topic_find_node ( topic_name );

	if ( _topic )
	{
		_topic->enable = enable ? 1 : 0;
		_res = 0;
	}

	_MSG_UNLOCK();

	return _res;
}

int msg_bus_init ( void )
{
	_msg_bus_check_init();

	return _msg_bus_init_state ? 0 : -1;
}

/* 导出到初始化列表 */
INIT_COMPONENT_EXPORT ( msg_bus_init );

/* 导出到 MSH 命令列表 */
MSH_CMD_EXPORT ( msg_bus_cmd, msg bus cmd );